package com.yanggu.flink.datastream_api.state.operatorstate

import cn.hutool.core.util.StrUtil
import org.apache.flink.api.common.state.MapStateDescriptor
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

import java.util.concurrent.TimeUnit
import scala.util.Random

/**
 * flink使用广播流和广播状态实现配置动态更新
 * https://blog.csdn.net/weixin_42155491/article/details/104886884
 */
object BroadcastStateDemo01 {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    //设置并行度为4
    env.setParallelism(4)

    //广播流只能使用MapState, 不支持其他状态, 例如ValueState和ListState等状态
    //定义MapState描述符
    //MapStateDescriptor的Key和Value的泛型如何确定呢？
    //key从主流和广播流中都能获取且是唯一标识
    //value表示对应关系, 一对一(对象)和一对多(Set[对象])
    val mapStateDescriptor = new MapStateDescriptor[String, String]("test broadcast map state", classOf[String], classOf[String])

    //自定义广播流 BroadcastStream
    val broadcastStream = env
      .addSource(new RuleSource())
      //广播流的并行度一般设置为1
      .setParallelism(1)
      //声明成广播流, 传入MapStateDescriptor
      .broadcast(mapStateDescriptor)

    //定义数据流
    val dataStream = env
      .addSource(new WordDataSource())
      .setParallelism(1)

    //数据流connect广播流
    dataStream
      .connect(broadcastStream)
      .process(new FilterKeyWordProcessFunction())
      .print("result>>>")

    env.execute()

  }

}

class RuleSource extends RichSourceFunction[String] {

  private var running = true

  //测试数据集
  private val dataSet = Array[String]("java", "swift", "php", "go", "python")

  override def open(parameters: Configuration): Unit = {
    //一般这里初始化数据集
    //例如建立和mysql的链接, 初始化数据
  }

  /**
   * 数据源：模拟每30秒随机更新一次拦截的关键字
   */
  override def run(ctx: SourceFunction.SourceContext[String]): Unit = {
    val length = dataSet.length
    val random = new Random
    while (running) {
      val str = dataSet(random.nextInt(length))
      ctx.collect(str)
      println(s"发送规则: $str")
      TimeUnit.SECONDS.sleep(30)
    }
  }

  override def cancel(): Unit = {
    running = false
  }

}

class WordDataSource extends RichSourceFunction[String] {

  private var running = true

  //测试数据集
  private val dataSet = Array[String]("java是世界上最优秀的语言",
    "swift是世界上最优秀的语言",
    "php是世界上最优秀的语言",
    "go是世界上最优秀的语言",
    "python是世界上最优秀的语言"
  )

  override def run(ctx: SourceFunction.SourceContext[String]): Unit = {
    val length = dataSet.length
    val random = new Random
    while (running) {
      val str = dataSet(random.nextInt(length))
      ctx.collect(str)
      println(s"向下游发送数据: $str")
      TimeUnit.SECONDS.sleep(3)
    }
  }

  override def cancel(): Unit = running = false

}

//三个泛型
//IN1 数据流的输入类型
//IN2 广播流的输入类型
//OUT 输出的数据类型
//这里需要继承BroadcastProcessFunction或者KeyedBroadcastProcessFunction抽象类重写两个抽象方法
class FilterKeyWordProcessFunction extends BroadcastProcessFunction[String, String, String] {

  private val key = "key"

  /**
   * 处理数据流输入的数据
   * 获取广播变量
   */
  override def processElement(value: String, ctx: BroadcastProcessFunction[String, String, String]#ReadOnlyContext, out: Collector[String]): Unit = {
    //这里的State值只读的, 在这里获取到广播变量后可以进行其他操作了, 广播变量一般都是作为配置一类的变量
    val mapStateDescriptor = new MapStateDescriptor[String, String]("test broadcast map state", classOf[String], classOf[String])
    val state = ctx.getBroadcastState(mapStateDescriptor)
    val str = state.get(key)
    if (StrUtil.isEmpty(str)) {
      return
    }
    //如果包含关键字向下游发送数据
    if (value.contains(str)) {
      out.collect(value)
    }
  }

  /**
   * 处理广播流数据的数据
   * 一般是更新广播变量中的数据
   */
  override def processBroadcastElement(value: String, ctx: BroadcastProcessFunction[String, String, String]#Context, out: Collector[String]): Unit = {
    //通过之前定义的MapStateDescriptor获取广播状态, 本质上是MapState
    val mapStateDescriptor = new MapStateDescriptor[String, String]("test broadcast map state", classOf[String], classOf[String])
    val state = ctx.getBroadcastState(mapStateDescriptor)
    //这里的MapState是可读和可写的
    state.put(key, value)
    println(s"关键字更新成功，更新拦截关键字: $value")
  }

}
